package com.cmc6.websocket.listener;

import cn.hutool.core.bean.BeanUtil;
import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.thread.ThreadUtil;
import cn.hutool.json.JSONObject;
import cn.hutool.json.JSONUtil;
import com.cmc6.common.constant.BaseConstant;
import com.cmc6.common.model.enums.SocketMessageEnum;
import com.cmc6.websocket.config.MyChannelHandlerPool;
import io.netty.channel.Channel;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import org.springframework.kafka.annotation.KafkaHandler;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import java.util.List;
import java.util.Set;

/**
 * socket 的kafka 监听器
 */
@Component
@KafkaListener(topics = {BaseConstant.KAFKA_SOCKET_TOPIC}, containerFactory = "dynamicGroupIdContainerFactory")
public class SocketKafkaListener {

    @KafkaHandler
    public void receive(String jsonStr) {
        // 异步处理消息
        ThreadUtil.execute(() -> {

            JSONObject jsonObject = JSONUtil.parseObj(jsonStr);

            Integer code = jsonObject.getInt("code");
            if (code == null) {
                return;
            }

            SocketMessageEnum socketMessageEnum = SocketMessageEnum.getByCode(code);
            if (socketMessageEnum == null) {
                return;
            }

            BeanUtil.copyProperties(jsonObject, socketMessageEnum);

            handleMessageEnum(socketMessageEnum);
        });
    }

    /**
     * 处理 socketMessageEnum
     */
    private void handleMessageEnum(SocketMessageEnum socketMessageEnum) {

        int code = socketMessageEnum.getCode();
        JSONObject json = socketMessageEnum.getJson();
        List<Channel> channelList;

        if (code == 1) {
            // socket连接记录主键id
        } else if (code >= 2 && code <= 4) {
            // 账号已在其他地方登录，您被迫下线
            // 登录过期，请重新登录
            // 账号已被注销
            Set<Number> idSet = json.get("userIdSet", Set.class);
            if (CollUtil.isEmpty(idSet)) {
                idSet = json.get("socketIdSet", Set.class);
                channelList = MyChannelHandlerPool.getChannelByIdSet(MyChannelHandlerPool.SOCKET_ID_KEY, idSet);
            } else {
                channelList = MyChannelHandlerPool.getChannelByIdSet(MyChannelHandlerPool.USER_ID_KEY, idSet);
            }

            writeAndFlush(channelList, socketMessageEnum);
        } else if (code == 5) {
            // 有新的通知
            Set<Number> userIdSet = json.get("userIdSet", Set.class);

            channelList = MyChannelHandlerPool.getChannelByIdSet(MyChannelHandlerPool.USER_ID_KEY, userIdSet);

            writeAndFlush(channelList, socketMessageEnum);
        } else if (code == 6) {

            // 有新的公告
            Set<Number> userIdSet = json.get("userIdSet", Set.class);

            if (CollUtil.isEmpty(userIdSet)) {
                writeAndFlush(null, socketMessageEnum);
            } else {
                channelList = MyChannelHandlerPool.getChannelByIdSet(MyChannelHandlerPool.USER_ID_KEY, userIdSet);
                writeAndFlush(channelList, socketMessageEnum);
            }
        }

    }

    /**
     * 推送消息
     */
    private void writeAndFlush(List<Channel> channelList, SocketMessageEnum socketMessageEnum) {

        socketMessageEnum.setJson(null);

        if (channelList == null) {

            MyChannelHandlerPool.send2All(socketMessageEnum); // 发送给所有人

        } else {
            TextWebSocketFrame textWebSocketFrame = socketMessageEnum.toSocketText();

            for (Channel item : channelList) {
                item.writeAndFlush(textWebSocketFrame);
            }
        }

    }

}
